package bma.common.netty.thrift.dual;

import java.util.List;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ExceptionEvent;

import bma.common.langutil.core.ExceptionUtil;
import bma.common.netty.NettyServer;
import bma.common.netty.handler.ChannelHandlerExceptionClose;
import bma.common.netty.handler.ChannelHandlerFramed;
import bma.common.netty.thrift.core.NettyThriftConfig;
import bma.common.netty.thrift.dual.impl.NTSDispatcherCommon;
import bma.common.netty.thrift.dual.impl.NTSModuleProcessor;

/**
 * 支持双向（上行，下行）的thrift协议通信
 * 
 * @author guanzhong
 * 
 */
public class NettyThriftDualServer extends NettyServer {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(NettyThriftDualServer.class);

	protected NettyThriftConfig config;
	protected NTSProcessor processor;

	public NettyThriftDualServer() {
		super();
	}

	public NTSProcessor getProcessor() {
		return processor;
	}

	public void setProcessor(NTSProcessor p) {
		this.processor = p;
	}

	public NettyThriftConfig getConfig() {
		return config;
	}

	public void setConfig(NettyThriftConfig config) {
		this.config = config;
	}

	public void setServices(List<NTSModuleProcessor> services) {
		if (this.processor != null) {
			throw new IllegalArgumentException("processor already set");
		}
		NTSDispatcherCommon dis = new NTSDispatcherCommon();
		dis.setServices(services);
		this.processor = dis;
	}

	@Override
	protected void beforeBuildPipeline(ChannelPipeline pipeline) {
		super.beforeBuildPipeline(pipeline);
		pipeline.addLast("framed",
				new ChannelHandlerFramed(config.getMaxLength()));
		config.sureTimeout();
		pipeline.addLast("service", new NCHThriftService(config, processor));
		pipeline.addLast("error", new ChannelHandlerExceptionClose() {
			@Override
			protected void logException(ChannelHandlerContext ctx,
					ExceptionEvent e) {
				if (log.isWarnEnabled()) {
					Throwable t = ExceptionUtil.cause(e.getCause());
					log.warn("{} error {}/{}", new Object[] {
							ctx.getChannel().getRemoteAddress(),
							t.getClass().getName(), t.getMessage() });
				}
			}
		});
	}

}
